package org.niit.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/*

   用Java 去生产数据 作为生产者
 */
public class KafkaProducerTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //1.配置信息 保存在Properties
        Properties props = new Properties();
            //连接地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        props.put(ProducerConfig.ACKS_CONFIG,"all");// 应答机制
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");//保证生产数据的一致性和不可重复性
        //序列号数据  我们生产的数据是字符串，Kafka所有的接收的数据是二进制，所有要字符串进行序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //2.创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<>(props);


        //3.生产并发送数据  生产1-100的数据 到 BD2
        for (int  i = 1; i<=100;i++){
            //构建一条数据
            ProducerRecord<String,String> record = new ProducerRecord<>("BD2",null,i+"");

            Future<RecordMetadata> res = producer.send(record);
            //发送成功的应答
            RecordMetadata metadata = res.get();//获得发送成功的元素 包含 主题 分区 偏移量 Key value 时间等数据
            System.out.println("第"+i+"条数据发送成功"+metadata.timestamp());

        }
        //4.关闭
        producer.close();

    }
}
